
package cn.tang.tframe.mq.producer;

import cn.tang.tframe.common.base.MainConfigLoader;
import cn.tang.tframe.common.utils.io.PropertiesUtil;
import cn.tang.tframe.common.utils.lang.SpiUtils;
import cn.tang.tframe.common.utils.lang.StringUtils;
import cn.tang.tframe.dynconfig.DynConfigRegister;
import cn.tang.tframe.dynconfig.IDynConfigProcessor;
import cn.tang.tframe.dynconfig.parser.ConfigValueParser;
import cn.tang.tframe.dynconfig.parser.StringConfigValueParser;
import cn.tang.tframe.mq.IQtMqFactory;
import cn.tang.tframe.mq.kafka.producer.KafkaProducerClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * Created by tanghc on 17/1/20.
 */
public class QtProducerClient {

	static Logger logger = LoggerFactory.getLogger(KafkaProducerClient.class);

	private QtProducerClient() {
	}

	private static class QtProducerClientHolder {

		private static IQtProducer instance;
		private static final String mq_config_group = "common";
		private static final String mq_config_dataId = "mqConfig";
		private static String mq = "kafka";
		private static boolean isInit = false;
		private static final String productCode =
			MainConfigLoader.getInstance().getProductCode();
		private static final String appCode =
			MainConfigLoader.getInstance().getAppCode();
		static {
			DynConfigRegister.getInstance().buildOnce(
				productCode, appCode, mq_config_group, mq_config_dataId,
				new IDynConfigProcessor() {

					@Override
					public void procesConfig(String config) {

						Properties properties =
							PropertiesUtil.loadFromString(config);
						IQtProducer iQtProducer = buildProducer(properties);
						if (null != iQtProducer) {
							instance = iQtProducer;
						}
						else {
							logger.warn(
								"didnot modify qtmq build client is null:productCode={},appCode={}",
								productCode, appCode);
						}
					}
				});
		}

		/**
		 */
		private static IQtProducer buildProducer(Properties properties) {

			if (null == properties) {
				properties = new Properties();
			}
			ConfigValueParser<String> configValueParser =
				new StringConfigValueParser();
			String newMq = configValueParser.parse(properties, "producer", "kafka");
			if (!isInit) {
				IQtMqFactory mqFactory =
					SpiUtils.getSpiBean(newMq, IQtMqFactory.class);
				if (null != mqFactory) {
					isInit = true;
					return mqFactory.builderPruduer();
				}
				else {
					throw new RuntimeException("no qtmqfactory spi:" + newMq);
				}
			}
			else {
				if (StringUtils.equalsIgnoreCase(newMq, mq)) {
					logger.info("not change mq spi do nothing");
				}
				else {
					IQtMqFactory mqFactory =
						SpiUtils.getSpiBean(newMq, IQtMqFactory.class);
					if (null != mqFactory) {
						return mqFactory.builderPruduer();
					}
					else {
						throw new RuntimeException(
							"no qtmqfactory spi:" + newMq);
					}
				}
			}
			return null;
		}
	}

	public static IQtProducer getQtProducer() {

		IQtProducer instance = QtProducerClientHolder.instance;
		if (null == instance) {
			throw new RuntimeException("build qtmq producer client exception");
		}
		return instance;
	}
}
